package com.atguigu.flink.sql.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/4/11
 *  sql中不支持计数窗口。
 *
 *      Group Window Functions: 常规的 window聚合
 *                                  提供了 滚动，会话，滑动。
 *
 *                   TVF window:  比Group Window更有优势
 *                          1.可以进行性能优化
 *                          2.支持标准的 GROUPING SETS
 *                          3.支持topN
 *
 *                  https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql/queries/window-tvf/
 *                          Tumble Windows
 *                          Hop Windows
 *                          Cumulate Windows:
 *                              https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql/queries/window-tvf/#cumulate
 *                          Session Windows (will be supported soon)
 *
 *   ------------------------
 *      会话窗口:  只能选 Group Window Functions
 *      滚动，滑动，累积:  TVF window
 *
 *    ------------------
 *      累积窗口:
 *          步长为 1 hour ，最大统计范围是  1 day  :
 *                  [00:00, 01:00),
 *                  [00:00, 01:00),[1:00,2:00)
 *                  [00:00, 01:00),[1:00,2:00),[2:00,3:00)
 *
 *                   …, [00:00, 24:00)
 *                   重置。
 *                    [00:00, 01:00),
 *                     ....
 *                     [00:00, 24:00)
 *
 *
 */
public class Demo4_GroupWindowSQL
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        //自带水印，自带eventtime
        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        Schema schema = Schema.newBuilder()
                             .column("id", "STRING")
                             .column("ts", "BIGINT")
                             .column("vc", "INT")
                             .columnByExpression("pt", "proctime()")
                             .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                              .watermark("et","et - INTERVAL '0.001' SECOND")
                             .build();
        //从流中获取时间属性
        Table table = tableEnvironment.fromDataStream(ds,schema);

        //为表起名字
        tableEnvironment.createTemporaryView("t1",table);

        String sessionWindowSql =
            "     select " +
            "        id," +
            "        SESSION_start( et ,INTERVAL '3' SECOND ) wstart," +
            "        SESSION_end( et ,INTERVAL '3' SECOND ) wend, " +
            "        sum(vc) sumVc " +
            "      from  t1 " +
            "      group by SESSION( et ,INTERVAL '3' SECOND ), id ";

        /*
            select from TABLE( 窗口函数()  ) GROUP BY window_start, window_end;

                TUMBLE(TABLE 表名, DESCRIPTOR(时间属性), INTERVAL '10' MINUTES): 滚动的窗口函数

                HOP(TABLE 表名, DESCRIPTOR(时间属性), slide  INTERVAL '5' MINUTES, size INTERVAL '10' MINUTES): 滑动的窗口函数
                    HOP table function based aggregate requires size must be an integral multiple of slide
                        tvf window，滑动窗口强制要求，size 必须是 slide的整倍数

                 CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), step INTERVAL '2' MINUTES, maxsize INTERVAL '10' MINUTES): 累积窗口

         */
        String cumulateWindowSql = "SELECT id, window_start, window_end, SUM(vc) sumVc" +
            "  FROM TABLE(" +
            "    CUMULATE(TABLE t1, DESCRIPTOR(et), INTERVAL '2' SECOND  , INTERVAL '6' SECOND))" +
            "  GROUP BY window_start, window_end , id";

        String tumbleWindowSql = "SELECT id, window_start, window_end, SUM(vc) sumVc" +
            "  FROM TABLE(" +
            "    TUMBLE(TABLE t1, DESCRIPTOR(et), INTERVAL '5' SECOND ))" +
            "  GROUP BY window_start, window_end , id";


        String hopWindowSql = "SELECT id, window_start, window_end, SUM(vc) sumVc" +
            "  FROM TABLE(" +
            "    HOP(TABLE t1, DESCRIPTOR(et),  INTERVAL '3' SECOND, INTERVAL '6' SECOND)" +
            "  )" +
            "  GROUP BY window_start, window_end , id";

        //使用sql操作
        tableEnvironment.sqlQuery(sessionWindowSql)
                        .execute().print();

        env.execute();


    }
}
